package com.hyf.kafka.demo;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class KafKaConsumer extends  Thread{

    private final ConsumerConnector consumerConnector;
    private final String topic;

    public KafKaConsumer(String topic) {
        this.topic = topic;
        consumerConnector=Consumer.createJavaConsumerConnector(createConsumerConfig());
    }
    private static ConsumerConfig createConsumerConfig(){
        Properties properties=new Properties();
        properties.put("zookeeper.connect",KafKaProperties.zkConnect);
        properties.put("group.id",KafKaProperties.groupId);
        properties.put("zookeeper.session.timeout.ms","40000");
        properties.put("zookeeper.sync.time.ms","200");
        properties.put("auto.commit.interval.ms","1000");

        return new ConsumerConfig(properties);
    }

    @Override
    public void run() {
        Map<String,Integer> topicCountMap=new HashMap<String, Integer>();
        topicCountMap.put(topic,new Integer(1));
        Map<String,List<KafkaStream<byte[],byte[]>>> consumerMap=consumerConnector.createMessageStreams(topicCountMap);
        KafkaStream<byte[],byte[]> stream=consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        while (it.hasNext()){
            System.out.println("receive:"+new String(it.next().message()));
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
